大志说编程

主题

09 结构化输出 ​

在前面的章节中，Agent返回的都是自然语言文本。但在实际开发中，我们往往需要Agent返回固定格式的数据，比如：

  • 从一段文本中提取联系人信息（姓名、邮箱、电话）
  • 分析一条商品评论（评分、情感、关键点）
  • 解析一段会议记录（任务、负责人、优先级）

如果Agent返回的是一段自然语言，你还得再写一套解析逻辑把它转成结构化数据，这就很麻烦了。结构化输出就是来解决这个问题的——直接让Agent返回你能用的数据格式。

一、快速上手 ​

使用create_agent的response_format参数，传入一个Pydantic模型作为输出格式，Agent就会自动返回结构化数据：

python
from pydantic import BaseModel, Field
from langchain.agents import create_agent


# 定义输出格式
class ContactInfo(BaseModel):
    """联系人信息"""
    name: str = Field(description="姓名")
    email: str = Field(description="邮箱地址")
    phone: str = Field(description="电话号码")


# 创建Agent时指定response_format
agent = create_agent(
    model="deepseek-v4-flash",
    response_format=ContactInfo,
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "从这段文本中提取联系人信息：张三，zhangsan@example.com，13800138000"}]
})

# 直接拿到结构化数据
print(result["structured_response"])
# ContactInfo(name='张三', email='zhangsan@example.com', phone='13800138000')

就这么简单。Agent会按照你定义的Pydantic模型来输出，返回的数据直接就是ContactInfo对象，你可以直接用result["structured_response"].name来访问字段。

二、支持的Schema类型 ​

除了Pydantic模型，还支持以下几种定义方式：

2.1 Pydantic模型（推荐） ​

最推荐的方式，因为Pydantic会帮你做数据校验，比如限制评分范围、必填字段等：

python
from pydantic import BaseModel, Field
from typing import Literal
from langchain.agents import create_agent


class ProductReview(BaseModel):
    """商品评论分析"""
    rating: int | None = Field(description="评分，1-5分", ge=1, le=5)
    sentiment: Literal["positive", "negative"] = Field(description="情感倾向")
    key_points: list[str] = Field(description="关键点，每个1-3个词")


agent = create_agent(
    model="deepseek-v4-flash",
    response_format=ProductReview,
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "分析这条评论：'很好的产品，5星好评，物流很快，就是有点贵'"}]
})

review = result["structured_response"]
print(f"评分: {review.rating}")
print(f"情感: {review.sentiment}")
print(f"关键点: {review.key_points}")

2.2 Dataclass ​

如果你不想引入Pydantic依赖，用Python自带的dataclass也行，但没有数据校验功能：

python
from dataclasses import dataclass
from langchain.agents import create_agent


@dataclass
class ContactInfo:
    """联系人信息"""
    name: str  # 姓名
    email: str  # 邮箱
    phone: str  # 电话


agent = create_agent(
    model="deepseek-v4-flash",
    response_format=ContactInfo,
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "提取联系人：李四，lisi@test.com，13900139000"}]
})

print(result["structured_response"])
# {'name': '李四', 'email': 'lisi@test.com', 'phone': '13900139000'}

2.3 TypedDict ​

和dataclass类似，返回的是字典：

python
from typing_extensions import TypedDict
from langchain.agents import create_agent


class ContactInfo(TypedDict):
    """联系人信息"""
    name: str  # 姓名
    email: str  # 邮箱
    phone: str  # 电话


agent = create_agent(
    model="deepseek-v4-flash",
    response_format=ContactInfo,
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "提取联系人：王五，wangwu@test.com，13700137000"}]
})

print(result["structured_response"])
# {'name': '王五', 'email': 'wangwu@test.com', 'phone': '13700137000'}

2.4 JSON Schema ​

如果你更习惯写JSON Schema，也可以直接传：

python
from langchain.agents import create_agent


contact_info_schema = {
    "type": "object",
    "description": "联系人信息",
    "properties": {
        "name": {"type": "string", "description": "姓名"},
        "email": {"type": "string", "description": "邮箱"},
        "phone": {"type": "string", "description": "电话"}
    },
    "required": ["name", "email", "phone"]
}

agent = create_agent(
    model="deepseek-v4-flash",
    response_format=contact_info_schema,
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "提取联系人：赵六，zhaoliu@test.com，13600136000"}]
})

print(result["structured_response"])
# {'name': '赵六', 'email': 'zhaoliu@test.com', 'phone': '13600136000'}

2.5 Union类型（多选一） ​

有时候你希望Agent根据输入内容自动选择最合适的输出格式。比如输入可能是商品评论，也可能是客户投诉，用Union类型可以让Agent自己判断：

python
from pydantic import BaseModel, Field
from typing import Literal, Union
from langchain.agents import create_agent


class ProductReview(BaseModel):
    """商品评论"""
    rating: int | None = Field(description="评分1-5", ge=1, le=5)
    sentiment: Literal["positive", "negative"] = Field(description="情感")
    key_points: list[str] = Field(description="关键点")


class CustomerComplaint(BaseModel):
    """客户投诉"""
    issue_type: Literal["product", "service", "shipping", "billing"] = Field(description="问题类型")
    severity: Literal["low", "medium", "high"] = Field(description="严重程度")
    description: str = Field(description="问题描述")


agent = create_agent(
    model="deepseek-v4-flash",
    response_format=Union[ProductReview, CustomerComplaint],
)

# 输入是评论，Agent会返回ProductReview
result = agent.invoke({
    "messages": [{"role": "user", "content": "分析：'很好的产品，5星好评'"}]
})
print(type(result["structured_response"]))  # ProductReview

# 输入是投诉，Agent会返回CustomerComplaint
result = agent.invoke({
    "messages": [{"role": "user", "content": "分析：'产品有质量问题，要求退款'"}]
})
print(type(result["structured_response"]))  # CustomerComplaint

三、两种实现策略 ​

LangChain内部有两种方式来实现结构化输出，一般情况下你不需要关心，但了解它们有助于排查问题。

3.1 Provider Strategy（模型原生） ​

有些模型本身支持结构化输出，比如OpenAI、Anthropic等。这种方式最可靠，因为是模型提供商在底层保证输出格式的正确性。

当你直接传一个Schema类型给response_format时，LangChain会自动判断：

  • 如果模型支持原生结构化输出 → 使用ProviderStrategy
  • 如果不支持 → 回退到ToolStrategy
python
# 直接传类型，LangChain自动选择策略
agent = create_agent(
    model="gpt-5.4",
    response_format=ContactInfo,
)

你也可以显式指定：

python
from langchain.agents.structured_output import ProviderStrategy

agent = create_agent(
    model="gpt-5.4",
    response_format=ProviderStrategy(ContactInfo),
)

3.2 Tool Strategy（工具调用） ​

对于不支持原生结构化输出的模型，LangChain通过工具调用来实现。原理是把你的Schema当作一个"虚拟工具"，让模型调用这个工具并传入参数，参数就是你要的结构化数据。

python
from langchain.agents.structured_output import ToolStrategy

agent = create_agent(
    model="deepseek-v4-flash",
    response_format=ToolStrategy(ContactInfo),
)

ToolStrategy有个好处是可以控制错误处理行为：

python
from langchain.agents.structured_output import ToolStrategy

# 默认行为：出错时自动重试
agent = create_agent(
    model="deepseek-v4-flash",
    response_format=ToolStrategy(ContactInfo, handle_errors=True),
)

# 自定义错误提示
agent = create_agent(
    model="deepseek-v4-flash",
    response_format=ToolStrategy(
        ContactInfo,
        handle_errors="请提供有效的联系人信息，包含姓名、邮箱和电话。"
    ),
)

# 关闭错误处理，直接抛异常
agent = create_agent(
    model="deepseek-v4-flash",
    response_format=ToolStrategy(ContactInfo, handle_errors=False),
)

四、错误处理 ​

模型在生成结构化输出时可能会犯错，比如：

  • 评分超出范围（给了10分，但限制是1-5）
  • 同时返回了多个结构化输出（Union类型时应该只返回一个）
  • 字段类型不对（应该传数字，传了字符串）

LangChain提供了自动重试机制。默认情况下handle_errors=True，当校验失败时，Agent会把错误信息反馈给模型，让它重新生成。

4.1 评分超出范围的例子 ​

python
from pydantic import BaseModel, Field
from langchain.agents import create_agent


class ProductRating(BaseModel):
    """商品评分"""
    rating: int | None = Field(description="评分1-5", ge=1, le=5)
    comment: str = Field(description="评论内容")


agent = create_agent(
    model="deepseek-v4-flash",
    response_format=ProductRating,
    system_prompt="你是一个评论解析助手，不要编造任何字段值。"
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "解析这条评论：'超级好的产品，10分好评！'"}]
})

print(result["structured_response"])
# 即使用户说了"10分"，模型也会自动纠正为5分以内
# ProductRating(rating=5, comment='超级好的产品')

整个过程是这样的：

  1. 模型第一次可能输出rating=10
  2. Pydantic校验失败，因为10 > 5
  3. LangChain把错误信息返回给模型："评分必须小于等于5"
  4. 模型重新生成，输出rating=5

4.2 自定义错误处理 ​

你可以用函数来自定义错误处理逻辑：

python
from langchain.agents.structured_output import ToolStrategy, StructuredOutputValidationError, MultipleStructuredOutputsError


def custom_error_handler(error: Exception) -> str:
    if isinstance(error, StructuredOutputValidationError):
        return "格式有问题，请检查后重新输出。"
    elif isinstance(error, MultipleStructuredOutputsError):
        return "你返回了多个结果，请只返回最相关的一个。"
    else:
        return f"出错了：{str(error)}"


agent = create_agent(
    model="deepseek-v4-flash",
    response_format=ToolStrategy(
        schema=Union[ContactInfo, EventDetails],
        handle_errors=custom_error_handler,
    ),
)

五、自定义工具消息 ​

使用ToolStrategy时，结构化输出会在对话历史中生成一条工具消息。默认消息是这样的：

Returning structured response: {'name': '张三', 'email': 'zhangsan@example.com', 'phone': '13800138000'}

你可以通过tool_message_content参数自定义这条消息：

python
from langchain.agents.structured_output import ToolStrategy


class MeetingAction(BaseModel):
    """会议待办事项"""
    task: str = Field(description="具体任务")
    assignee: str = Field(description="负责人")
    priority: Literal["low", "medium", "high"] = Field(description="优先级")


agent = create_agent(
    model="deepseek-v4-flash",
    response_format=ToolStrategy(
        schema=MeetingAction,
        tool_message_content="待办事项已记录！"
    ),
)

这样在对话历史中，工具消息就会显示"待办事项已记录！"而不是原始的JSON数据。

六、总结 ​

结构化输出让Agent返回可直接使用的数据，而不是自然语言文本。核心要点：

  • 使用response_format参数指定输出格式，推荐用Pydantic模型
  • 结果通过result["structured_response"]获取
  • LangChain会自动选择最优的实现策略（原生支持用ProviderStrategy，否则用ToolStrategy）
  • 内置错误重试机制，校验失败时会自动让模型重新生成

在下一篇文章中，我们将学习LangChain的中间件机制，它是扩展Agent功能的重要方式。